package exlocalsubpub

import (
	"fmt"
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/subpub"
	"sync"
	"sync/atomic"
	"time"
)

type exMsgItem struct {
	wg    sync.WaitGroup
	topic string
	max   int
	args  []interface{}
	ret   int
}

type ExLocalNode struct {
	closedflag  int32
	startedflag int32
	step        int8
	lk          sync.RWMutex
	que         int8
	timeout     time.Duration
	msgQue      chan *exMsgItem
	ex          subpub.ISubPub
	pubFunc     func(topic string, max int, args ...interface{}) int
}

func (this *ExLocalNode) checkStart() error {
	this.lk.Lock()
	defer this.lk.Unlock()
	if this.startedflag == 1 {
		return nil
	}
	defer func() {
		this.startedflag = 1
	}()
	if this.que == 1 {
		go this.innerProcessQueue()
	}
	return nil
}

func (this *ExLocalNode) innerProcessCmd(cmd *exMsgItem) {
	cmd.ret = this.ex.Pub(cmd.topic, cmd.max, cmd.args...)
	cmd.wg.Done()
}

func (this *ExLocalNode) innerProcessQueue() {
	if gobase.GoFunCatchException {
		defer gobase.DeferCatchPanic()
	}

	this.step = 1
	defer func() {
		this.step = 9
	}()

	ticker := time.NewTicker(time.Millisecond * 100)
	for atomic.LoadInt32(&this.closedflag) == 0 {
		this.step = 2
		select {
		case cmd := <-this.msgQue:
			if cmd == nil {
				break
			}
			this.step = 3
			this.innerProcessCmd(cmd)
		case <-ticker.C:
			this.step = -1

		}
	}
}

func (this *ExLocalNode) innerPubQue(topic string, max int, args ...interface{}) int {
	itm := &exMsgItem{topic: topic, max: max, args: args}
	itm.wg.Add(1)
	select {
	case this.msgQue <- itm:
	case <-time.After(this.timeout):
		itm.wg.Done()
		return -1
	}
	itm.wg.Wait()
	return 0
}

func (this *ExLocalNode) Pub(topic string, max int, args ...interface{}) int {
	return this.pubFunc(topic, max, args...)
}

func (this *ExLocalNode) Close() error {
	select {
	case this.msgQue <- nil:
	case <-time.After(time.Second * 3):
		close(this.msgQue)
	}
	atomic.StoreInt32(&this.closedflag, 1)
	return nil
}

var (
	DefaultExGroup = gobase.NewConnectObjectRefStorage(newExNode)
	lstEx          gobase.SyncMap
)

func checkGetEx(conf gobase.StrMap) subpub.ISubPub {
	extype := conf.IntByName("extype", 0)
	idstr := fmt.Sprintf("%d", extype)
	return lstEx.LoadOrStoreFunc(idstr, func() interface{} {
		if extype == 0 {
			return subpub.NewSubchannel()
		} else {
			return subpub.NewSubscribe()
		}
	}, nil).(subpub.ISubPub)
}

// extype=0&que=1&timeout=1000
func newExNode(connstr string) interface{} {
	conf := gobase.NewStrMap()
	conf.URLFormDecode(connstr)
	node := &ExLocalNode{}
	node.que = conf.Int8ByName("que", 1)
	node.ex = checkGetEx(conf)
	if node.que == 0 {
		node.pubFunc = node.ex.Pub
	} else {
		node.msgQue = make(chan *exMsgItem)
		node.pubFunc = node.innerPubQue
		node.timeout = time.Duration(conf.IntByName("timeout", 1000)) * time.Millisecond
	}
	return node
}
